package com.joizhang.seckill.kafka;

import com.joizhang.seckill.api.kafka.SecKillMessage;
import com.joizhang.seckill.utils.BeanUtil;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Component;
import org.springframework.util.concurrent.ListenableFuture;

import java.util.concurrent.ExecutionException;

@Slf4j
@Component
public class MQSender {

    private final KafkaTemplate<String, String> kafkaTemplate;

    @Autowired
    public MQSender(KafkaTemplate<String, String> kafkaTemplate) {
        this.kafkaTemplate = kafkaTemplate;
    }

    public void send(Object message) {
        String msg = BeanUtil.beanToString(message);
        log.info("send message: " + msg);
        kafkaTemplate.send(MQConfig.TEST_QUEUE, msg);
    }

    public Boolean sendSecKillMessage(SecKillMessage message) {
        String msg = BeanUtil.beanToString(message);
        // log.info("send sec-killing message: " + msg);
        kafkaTemplate.send(MQConfig.SECKILL_QUEUE, msg);
        return true;
    }

    public Boolean sendSecKillMessageSync(SecKillMessage message) {
        String msg = BeanUtil.beanToString(message);
         log.info("send sec-killing message: " + msg);
        ListenableFuture<SendResult<String, String>> sendResultListenableFuture =
                kafkaTemplate.send(MQConfig.SECKILL_QUEUE, msg);
        try {
            SendResult<String, String> stringStringSendResult = sendResultListenableFuture.get();
            log.info(stringStringSendResult.toString());
            return true;
        } catch (InterruptedException | ExecutionException e) {
            log.error(e.toString());
            return false;
        }
    }
}

